package dal

import (
	"context"
	"database/sql"
	"fmt"
	"strconv"
	"time"

	"gorm.io/plugin/dbresolver"

	"strings"

	"code.byted.org/bytertc/whiteboard_server/biz/dal/ddl"
	"code.byted.org/bytertc/whiteboard_server/biz/dal/spec"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/biz_error"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/kv"
	"code.byted.org/bytertc/whiteboard_server/biz/service/action"
	"code.byted.org/gopkg/logs/v2/log"
	"github.com/pkg/errors"
	"github.com/spf13/cast"
)

// ForMaxCycleNum for循环最大循环次数，防止死循环
const ForMaxCycleNum = 10000

// PerPageNum zset 分页读取每页数量
const PerPageNum = 100

// LockObjExpire 冲突对象锁定最长时间：2m10s
const LockObjExpire = 130

// SplitCharacters 切割分隔符
const SplitCharacters = "$^*%$"

// zset, member = object_id
func genLockSetKey(unique spec.BoardUnique) string {
	return fmt.Sprintf("locks:%s:%s:%d", unique.GetAppId(), unique.GetRoomId(), unique.GetBoardId())
}

func genLockSetMember(action action.Action) string {
	return fmt.Sprintf("%s%s%s", action.GetObjId(), SplitCharacters, action.GetOperator())
}

func genLockKey(action action.Action) string {
	return fmt.Sprintf("objLocks:%s:%s:%d:%s", action.GetAppId(), action.GetRoomId(), action.GetBoardId(), action.GetObjId())
}

func genLockVal(action action.Action) string {
	return action.GetOperator()
}

func genLockAudioKey(page spec.PageUnique) string {
	return fmt.Sprintf("page_audio:%s:%s:%d:%s", page.GetAppId(), page.GetRoomId(), page.GetBoardId(), page.GetPageId())
}

func genLockVideoKey(page spec.PageUnique) string {
	return fmt.Sprintf("page_video:%s:%s:%d:%s", page.GetAppId(), page.GetRoomId(), page.GetBoardId(), page.GetPageId())
}

func genHeratbeatKey(unique spec.RoomUnique) string {
	return fmt.Sprintf("heartbeatStamp:%s:%s", unique.GetAppId(), unique.GetRoomId())
}

// string, value = max packet id of operator
func genPacketIdKey(unique spec.OperatorUnique) string {
	return fmt.Sprintf("pktId:%s:%s:%s", unique.GetAppId(), unique.GetRoomId(), unique.GetOperator())
}

// 鉴权，校验自己是否有权限操作该对象
func (r *repo) Authenticate(ctx context.Context, action action.Action) error {
	lockKey := genLockKey(action)
	lockVal := genLockVal(action)
	permissionVal, err := r.cache.Get(ctx, lockKey)
	if err != nil && !IsNotFoundError(err) {
		return err
	}
	if permissionVal == lockVal {
		return nil
	}
	err = errors.Errorf("Authenticate fail for obj lock key %s with value %s,permissionVal:%s", lockKey, lockVal, permissionVal)
	log.V1.CtxError(ctx, "Authenticate fail %+v", err)
	return biz_error.Conflict.Wrap(err)
}

func (r *repo) LockObject(ctx context.Context, action action.Action) (err error) {
	defer func() {
		if err == nil {
			entry := kv.ZMember{
				Score:  cast.ToFloat64(time.Now().Unix()),
				Member: genLockSetMember(action),
			}
			r.cache.ZAdd(ctx, genLockSetKey(action), entry)
			r.cache.Expire(ctx, genLockSetKey(action), time.Hour)
		}
	}()
	lockKey := genLockKey(action)
	lockVal := genLockVal(action)
	expire := time.Second * 10
	ret, err := r.cache.CompareAndSet(ctx, lockKey, lockVal, lockVal, expire)
	if err != nil {
		return err
	}
	switch ret {
	case kv.Success:
		return nil
	case kv.NotExists:
		ok, err := r.cache.SetNX(ctx, lockKey, lockVal, expire)
		if err != nil {
			return biz_error.CacheError.Wrap(errors.WithStack(err))
		}
		if ok {
			return nil
		} else {
			//校验心跳保活和action同时触发setnx
			val, _ := r.cache.Get(ctx, lockKey)
			if val == lockVal {
				return nil
			}
		}
	case kv.Fail:
	}
	return biz_error.Conflict.Wrap(errors.Errorf("fail to lock key %s with value %s", lockKey, lockVal))
}
func (r *repo) UnlockObject(ctx context.Context, action action.Action) error {
	_, err := r.cache.CompareAndDelete(ctx, genLockKey(action), genLockVal(action))
	if err != nil {
		return err
	}
	r.cache.ZRem(ctx, genLockSetKey(action), genLockSetMember(action))

	return err
}

func (r *repo) GetLockedObjects(ctx context.Context, unique spec.BoardUnique) (map[string]string, error) {
	beginTime := cast.ToFloat64(time.Now().Unix() - LockObjExpire)
	lockMaps := make(map[string]string)
	for i := int64(0); i < ForMaxCycleNum; i++ {
		start := i * PerPageNum
		end := (i+1)*PerPageNum - 1
		z, err := r.cache.ZRangeWithScore(ctx, genLockSetKey(unique), start, end)
		if err != nil {
			return nil, biz_error.CacheError.Wrap(errors.WithStack(err))
		}
		for _, entry := range z {
			if entry.Score >= beginTime {
				parts := strings.Split(cast.ToString(entry.Member), SplitCharacters)
				if len(parts) == 0 {
					continue
				}
				if len(parts) == 1 {
					continue
				}
				lockMaps[parts[0]] = parts[1]
			}
		}
		if len(z) < PerPageNum {
			break
		}

	}
	return lockMaps, nil
}

// GetActionCount 获取单页内所有 object 各自的 action count
func (r *repo) GetActionCount(ctx context.Context, unique spec.PageUnique) (map[string]int32, error) {
	var cursor uint64
	var err error
	var entrys []string
	result := make(map[string]int32)
	for {
		entrys, cursor, err = r.cache.HScan(ctx, genActionCountKey(unique), "", cursor, 100)
		if err != nil {
			return nil, biz_error.CacheError.Wrap(errors.WithStack(err))
		}
		for i := 0; i < len(entrys)-1; i += 2 {
			field := cast.ToString(entrys[i])
			count := cast.ToInt32(entrys[i+1])
			if count == 0 {
				continue
			}
			result[field] = count
		}
		if cursor == 0 {
			break
		}
	}
	if len(result) == 0 {
		return r.refreshActionCount(ctx, unique)
	}
	if err != nil {
		return nil, biz_error.CacheError.Wrap(errors.WithStack(err))
	}
	return result, nil
}

// RefreshActionCount 刷新 action count 的统计
func (r *repo) refreshActionCount(ctx context.Context, unique spec.PageUnique) (map[string]int32, error) {
	var counts []struct {
		ObjectId    string `gorm:"column:object_id"`
		ActionCount int32  `gorm:"column:action_count"`
	}

	err := r.db.WithContext(ctx).Clauses(dbresolver.Write).Table(ddl.TableObjects).
		Where("app_id = ? and room_id = ? and board_id = ? and page_id = ? and initialized = true and deleted_at is null",
			unique.GetAppId(), unique.GetRoomId(), unique.GetBoardId(), unique.GetPageId()).
		Select("object_id", "action_count").Find(&counts).Error
	if err != nil {
		return nil, biz_error.DatabaseError.Wrap(errors.WithStack(err))
	}

	rawInput := make(map[string]interface{}, len(counts))
	countMap := make(map[string]int32, len(counts))
	for _, value := range counts {
		rawInput[value.ObjectId] = value.ActionCount
		countMap[value.ObjectId] = value.ActionCount
	}
	if len(countMap) == 0 {
		return countMap, nil
	}
	countKey := genActionCountKey(unique)
	pipe := r.cache.NewPipeline(ctx)
	pipe.HMSet(countKey, rawInput)
	pipe.Expire(countKey, time.Hour*24)
	_, err = pipe.Exec()
	//执行失败，则缩短过期时间，让后面请求重新覆盖数据
	if err != nil {
		r.cache.Expire(ctx, countKey, time.Second*5)
	}
	return countMap, biz_error.CacheError.Wrap(errors.WithStack(err))
}

// GetPageActionCount 获取单页内所有 object 的 action count 总数
func (r *repo) GetPageActionCount(ctx context.Context, unique spec.PageUnique) (int32, error) {
	actionCountKey := genActionCountKey(unique)
	var cursor uint64
	var actionCount int32
	var err error
	var entrys []string
	defer func() {
		if err != nil {
			err = errors.WithMessagef(err, "singleflight.Do GetActionCount fail appId:%s,roomId:%s,boardId:%d,pageId:%s", unique.GetAppId(), unique.GetRoomId(), unique.GetBoardId(), unique.GetPageId())
		}
	}()
	r.duplicateDelObject(ctx, unique)
	for {
		entrys, cursor, err = r.cache.HScan(ctx, actionCountKey, "", cursor, 100)
		if err != nil {
			return 0, biz_error.CacheError.Wrap(errors.WithStack(err))
		}
		for idx, val := range entrys {
			if idx%2 != 0 {
				actionCount += cast.ToInt32(val)
			}
		}
		if cursor == 0 {
			break
		}
	}
	if len(entrys) > 0 || actionCount > 0 {
		return actionCount, nil
	}
	count, err, _ := MSingleflight.Do(actionCountKey, func() (interface{}, error) {
		objects, err := r.getActionCountFromDB(ctx, unique)
		if err != nil {
			return 0, err
		}
		count, err := r.reBuildActionCount2Cache(ctx, unique, objects)
		return count, err

	})

	return cast.ToInt32(count), err

}

func (r *repo) getActionCountFromDB(ctx context.Context, unique spec.PageUnique) ([]ddl.Objects, error) {
	var objs []ddl.Objects
	err := r.db.WithContext(ctx).Clauses(dbresolver.Write).Table(ddl.TableObjects).
		Where(`app_id = ? and room_id = ? and board_id = ? and page_id = ? and initialized = true and deleted_at is null`,
			unique.GetAppId(), unique.GetRoomId(), unique.GetBoardId(), unique.GetPageId()).
		Select("object_id", "action_count").Find(&objs).Error
	if errors.Is(err, sql.ErrNoRows) {
		return objs, nil
	}
	if err != nil {
		return nil, biz_error.DatabaseError.Wrap(errors.WithStack(err))
	}
	return objs, nil

}

func (r *repo) reBuildActionCount2Cache(ctx context.Context, unique spec.PageUnique, objs []ddl.Objects) (totalCount int32, err error) {
	rawInput := make(map[string]interface{}, len(objs))
	for _, value := range objs {
		rawInput[value.ObjectId] = value.ActionCount
		totalCount += value.ActionCount
	}
	//防止穿透
	if len(rawInput) == 0 {
		rawInput["reWriteFlag"] = 0
	}
	countKey := genActionCountKey(unique)
	pipe := r.cache.NewPipeline(ctx)
	pipe.HMSet(countKey, rawInput)
	pipe.Expire(countKey, time.Hour*24)
	_, err = pipe.Exec()
	if err != nil {
		r.cache.Expire(ctx, countKey, time.Second*5)
		err = biz_error.CacheError.Wrap(errors.WithStack(err))
	}
	return

}

// GetPageControlCount 获取某一页的 control count
func (r *repo) GetPageControlCount(ctx context.Context, unique spec.PageUnique) (int32, error) {
	controlCount := sql.NullInt64{}
	err := r.db.WithContext(ctx).Table(ddl.TablePages).Where(
		"app_id = ? and room_id = ? and board_id = ? and page_id = ?",
		unique.GetAppId(), unique.GetRoomId(), unique.GetBoardId(), unique.GetPageId()).
		Select("control_count").Row().Scan(&controlCount)
	if errors.Is(err, sql.ErrNoRows) {
		return 0, nil
	}
	return int32(controlCount.Int64), biz_error.DatabaseError.Wrap(errors.WithStack(err))
}

func (r *repo) GetBoardControlCount(ctx context.Context, unique spec.BoardUnique) (int32, error) {
	controlCount := sql.NullInt64{}
	err := r.db.WithContext(ctx).Table(ddl.TablePages).Where(
		"app_id = ? and room_id = ? and board_id = ? and deleted_at is null",
		unique.GetAppId(), unique.GetRoomId(), unique.GetBoardId()).
		Select("sum(control_count)").Row().Scan(&controlCount)
	if errors.Is(err, sql.ErrNoRows) {
		return 0, nil
	}
	return int32(controlCount.Int64), biz_error.DatabaseError.Wrap(errors.WithStack(err))
}

func (r *repo) UpdatePacketId(ctx context.Context, unique spec.OperatorUnique, currentPacketId int32) error {
	expectedOldValue := strconv.FormatInt(int64(currentPacketId-1), 10)
	newValue := strconv.FormatInt(int64(currentPacketId), 10)
	key := genPacketIdKey(unique)
	ret, err := r.cache.CompareAndSet(ctx, key, expectedOldValue, newValue, time.Hour*24)
	if err != nil {
		return biz_error.CacheError.Wrap(errors.WithStack(err))
	}
	switch ret {
	case kv.NotExists:
		_, err = r.cache.Set(ctx, key, newValue, time.Hour*24)
		return biz_error.CacheError.Wrap(errors.WithStack(err))
	case kv.Fail:
		return biz_error.PacketReject.Wrap(errors.Errorf("fail to check packet id %d", currentPacketId))
	case kv.Success:
	}
	return nil
}

func (r *repo) GetPacketId(ctx context.Context, unique spec.OperatorUnique) (int32, error) {
	value, err := r.cache.Get(ctx, genPacketIdKey(unique))
	if err != nil {
		if IsNotFoundError(err) {
			return 0, nil
		}
		return 0, biz_error.CacheError.Wrap(errors.WithStack(err))
	}
	packetId, err := strconv.Atoi(value)
	return int32(packetId), biz_error.DecodeError.Wrap(errors.WithStack(err))
}

// 二次删除
func (r *repo) duplicateDelObject(ctx context.Context, unique spec.PageUnique) {
	//防止提前删除了，又被添加上
	safetyThreshold := int64(3)
	endTime := cast.ToFloat64(time.Now().Unix() - safetyThreshold)
	delObjectListKey := genDelObjectListKey(unique)
	//只取有限数量的
	zMembers, err := r.cache.ZRangeWithScore(ctx, delObjectListKey, 0, 100)
	objectList := make([]string, 0, len(zMembers))
	for _, entry := range zMembers {
		if entry.Score > endTime {
			break
		}
		objectList = append(objectList, cast.ToString(entry.Member))
	}
	if len(objectList) == 0 {
		return
	}
	actionCountKey := genActionCountKey(unique)
	pip := r.cache.NewPipeline(ctx)
	for _, object := range objectList {
		pip.HDel(actionCountKey, object)
	}
	cmds, _ := pip.Exec()
	//log.V1.CtxInfo(ctx, "cmd:%f+v", cmds)
	deletedObjects := make([]interface{}, 0)
	for idx, cmd := range cmds {
		if cmd.Err() != nil {
			log.V1.CtxError(ctx, "pip.HDel key:%s fail,objectId:%s,err:%+v", actionCountKey, objectList[idx], err)
		} else {
			deletedObjects = append(deletedObjects, objectList[idx])
		}
	}
	if len(deletedObjects) > 0 {
		err = r.cache.ZRem(ctx, delObjectListKey, deletedObjects...)
		if err != nil {
			log.V1.CtxError(ctx, "ZRem key:%s fail Objects:%+v,err:%+v", delObjectListKey, deletedObjects, err)
		}
	}
}

func (r *repo) SetHeartbeatStamp(ctx context.Context, unique spec.RoomUnique) (bool, error) {
	return r.cache.SetNX(ctx, genHeratbeatKey(unique), "1", time.Second*5)
}
